package pers.cz.redis;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import javax.annotation.PostConstruct;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * redis消息队列消费者模型: 可以处理文件上传后续处理等异步操作
 * @program: PostGirl-panent
 * @description: RedisQueueCustomer
 * @author: Cheng Zhi
 * @create: 2021-07-03 14:28
 **/
@Service
public class RedisQueueCustomer{

    @Autowired
    JedisPool jedisPool;

    @Value("${redis.channel}")
    private String channals; // 从配置文件中获取订阅的渠道

    // 初始化核心线程数
    private Integer threadPoolSize = 2;
    // 初始化最大线程数
    private Integer maxThreadPoolSize = 2;
    // 初始化空闲线程存存活时间
    private Long survivalTime = 0L;
    // 线程池任务队列大小
    private Integer threadPoolQueueSize = 20;
    // Rediis中List实现队列，使用brpop设置timeOut=0表示启用阻塞
    private Integer timeOut = 0;

    // 定义处理定时任务的线程数
    private final Integer TestStartThreadCount = 3;

    private final Integer InterfaceSyncThreadCount = 20;

    @PostConstruct
    public void init() {
        dealThread();
    }

   private void dealThread() {

       // 为每个需要订阅的渠道创建一个监听线程
       String[] channalsArray = channals.split(",");

       this.threadPoolSize = channalsArray.length + TestStartThreadCount + InterfaceSyncThreadCount;
       this.maxThreadPoolSize = channalsArray.length + TestStartThreadCount + InterfaceSyncThreadCount;
       // 创建一个线程池
       ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
               this.threadPoolSize,
               this.maxThreadPoolSize,
               this.survivalTime,
               TimeUnit.MILLISECONDS,
               new ArrayBlockingQueue<Runnable>(this.threadPoolQueueSize));
       // 如果线程池任务队列已满，则由主线程来执行这个任务
       threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

       // 处理订阅发布模式下的队列
       for (String channel : channalsArray) {
           // 获取redis连接对象
           Jedis resource = jedisPool.getResource();
           Jedis resourceDtl = jedisPool.getResource();
           MessageQueueDealThread thread = new MessageQueueDealThread(channel, resource,resourceDtl);
           threadPoolExecutor.execute(thread);
       }

       // 处理消费者模式下的队列
        for (int i=0; i<3; i++) {
            Jedis resource = jedisPool.getResource();
            TestCaseDealThread jobDealThread = new TestCaseDealThread(resource);
            threadPoolExecutor.execute(jobDealThread);
        }

        // 处理接口录制队列
       for (int i=0; i<3; i++) {
           Jedis resource = jedisPool.getResource();
           InterfaceSyncThread interfaceSyncThread = new InterfaceSyncThread(resource);
           threadPoolExecutor.execute(interfaceSyncThread);
       }

       // 处理sql录制队列
       for (int i=0; i<3; i++) {
           Jedis resource = jedisPool.getResource();
           InterfaceSqlSyncThread interfaceSqlSyncThread = new InterfaceSqlSyncThread(resource);
           threadPoolExecutor.execute(interfaceSqlSyncThread);
       }

       // 处理方法录制
       for (int i=0; i<3; i++) {
           Jedis resource = jedisPool.getResource();
           InterfaceMethodParamsSyncThread interfaceMethodParamsSyncThread = new InterfaceMethodParamsSyncThread(resource);
           threadPoolExecutor.execute(interfaceMethodParamsSyncThread);
       }

   }

}


